package com.gongfutrip.trans.remoting;

import java.net.SocketAddress;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.gongfutrip.trans.remoting.common.Pair;
import com.gongfutrip.trans.remoting.common.RemotingHelper;
import com.gongfutrip.trans.remoting.common.SemaphoreReleaseOnlyOnce;
import com.gongfutrip.trans.remoting.common.ServiceThread;
import com.gongfutrip.trans.remoting.expception.RemotingSendRequestException;
import com.gongfutrip.trans.remoting.expception.RemotingTimeoutException;
import com.gongfutrip.trans.remoting.expception.RemotingTooMuchRequestException;
import com.gongfutrip.trans.remoting.netty.NettyEvent;
import com.gongfutrip.trans.remoting.netty.ResponseFuture;
import com.gongfutrip.trans.remoting.protocol.RemotingCommand;
import com.gongfutrip.trans.remoting.protocol.RemotingSysResponseCode;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;

public abstract class NettyRemotingAbstract {

    private static final Logger logger = LoggerFactory.getLogger(NettyRemotingAbstract.class);

    private final Semaphore semaphoreOneway;

    private final Semaphore semaphoreAsync;

    protected Pair<NettyRequestProcessor, ExecutorService> defaultRequestProcessor;

    private static volatile int configVersion = -1;

    /* request code -->Pair */
    protected final HashMap<Integer, Pair<NettyRequestProcessor, ExecutorService>> processorTable =
        new HashMap<>(64);

    /* request code -->ResponseFuture */
    protected final ConcurrentHashMap<Integer, ResponseFuture> responseTable =
        new ConcurrentHashMap<>(256);

    protected final NettyEventExecuter nettyEventExecuter = new NettyEventExecuter();

    /**
     * 设置可执行线程数
     */
    public NettyRemotingAbstract(final int permitsOneway, final int permitsAsync) {
        this.semaphoreOneway = new Semaphore(permitsOneway, true);
        this.semaphoreAsync = new Semaphore(permitsAsync, true);
    }

    public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg)
        throws Exception {
        final RemotingCommand cmd = msg;
        if (cmd != null) {
            switch (cmd.getType()) {
                case REQUEST_COMMAND:
                    processRequestCommand(ctx, cmd);// 服务端处理从客户端收到的请求
                    break;
                case RESPONSE_COMMAND:
                    processResponseCommand(ctx, cmd);// 客户端处理从服务端发送过来的回应
                    break;
                default:
                    break;
            }
        }
    }

    public abstract RPCHook getRPCHook();

    public abstract ChannelEventListener getChannelEventListener();

    public void processResponseCommand(ChannelHandlerContext ctx, RemotingCommand cmd) {
        final int opaque = cmd.getOpaque();
        final ResponseFuture responseFuture = responseTable.get(opaque);
        if (responseFuture != null) {
            responseFuture.setResponseCommand(cmd);

            responseFuture.release();

            responseTable.remove(opaque);
            if (responseFuture.getInvokeCallback() != null) {// 收到消息之后异步回调
                executeInvokeCallback(responseFuture);
            } else {
                responseFuture.putResponse(cmd);
            }
        } else {
            logger.warn("receive response, but not matched any request, "
                + RemotingHelper.parseChannelRemoteAddr(ctx.channel()));
            logger.warn(cmd.toString());
        }
    }

    public void scanResponseTable() {
        final List<ResponseFuture> rfList = new LinkedList<ResponseFuture>();
        Iterator<Entry<Integer, ResponseFuture>> it = this.responseTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<Integer, ResponseFuture> next = it.next();
            ResponseFuture rep = next.getValue();
            // 超时
            if ((rep.getBeginTimestamp() + rep.getTimeoutMillis() + 1000) <= System
                .currentTimeMillis()) {// 超时移除
                rep.release();
                it.remove();
                rfList.add(rep);
                logger.warn("remove timeout request, " + rep);
            }
        }

        for (ResponseFuture responseFuture : rfList) {// 异步通知超时
            executeInvokeCallback(responseFuture);
        }
    }

    /**
     * 在线程池执行回调函数，如果线程池为空，则在当前线程执行
     */
    private void executeInvokeCallback(final ResponseFuture responseFuture) {
        ExecutorService executor = this.getCallbackExecutor();
        if (null != executor) {
            try {
                executor.submit(() -> responseFuture.executeInvokeCallback());
            } catch (Throwable e) {
                logger.warn("execute callback in executor exception, and callback throw", e);
            }
        } else {
            try {
                responseFuture.executeInvokeCallback();
            } catch (Throwable e) {
                logger.warn("executeInvokeCallback Exception", e);
            }
        }

    }

    public abstract ExecutorService getCallbackExecutor();

    public void putNettyEvent(final NettyEvent event) {
        this.nettyEventExecuter.putNettyEvent(event);
    }

    public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
        final long timeoutMillis)
        throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
        final int opaque = request.getOpaque();
        try {
            final ResponseFuture responseFuture = new ResponseFuture(opaque, timeoutMillis, null,
                null);
            this.responseTable.put(opaque, responseFuture);
            final SocketAddress addr = channel.remoteAddress();
            channel.writeAndFlush(request).addListener(new ChannelFutureListener() {

                public void operationComplete(ChannelFuture f) throws Exception {
                    if (f.isSuccess()) {
                        responseFuture.setSendRequestOK(true);
                        return;
                    } else {
                        responseFuture.setSendRequestOK(false);
                        responseTable.remove(opaque);
                        responseFuture.setCause(f.cause());
                        responseFuture.putResponse(null);
                        logger.warn("send a request command to channel <" + addr + "> failed.");
                    }
                }
            });
            RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
            if (null == responseCommand) {
                if (responseFuture.isSendRequestOK()) {
                    throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr),
                        timeoutMillis, responseFuture.getCause());
                } else {
                    throw new RemotingSendRequestException(
                        RemotingHelper.parseSocketAddressAddr(addr),
                        responseFuture.getCause());
                }
            }
            return responseCommand;
        } finally {
            this.responseTable.remove(opaque);
        }
    }

    public void invokeOnewayImpl(final Channel channel, final RemotingCommand request,
        final long timeoutMillis) throws InterruptedException, RemotingTooMuchRequestException,
        RemotingTimeoutException, RemotingSendRequestException {
        request.markOnewayRPC();
        boolean acquired = this.semaphoreOneway.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        if (acquired) {
            final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(
                this.semaphoreOneway);
            try {
                channel.writeAndFlush(request).addListener(new ChannelFutureListener() {

                    public void operationComplete(ChannelFuture f) throws Exception {
                        once.release();
                        if (!f.isSuccess()) {
                            logger.warn(
                                "send a request command to channel <" + channel.remoteAddress()
                                    + "> failed.");
                        }
                    }
                });
            } catch (Exception e) {
                once.release();
                logger.warn(
                    "write send a request command to channel <" + channel.remoteAddress()
                        + "> failed.");
                throw new RemotingSendRequestException(
                    RemotingHelper.parseChannelRemoteAddr(channel), e);
            }
        } else {
            if (timeoutMillis <= 0) {
                throw new RemotingTooMuchRequestException("invokeOnewayImpl invoke too fast");
            } else {
                String info = String.format(
                    "invokeOnewayImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
                    //
                    timeoutMillis, //
                    this.semaphoreOneway.getQueueLength(), //
                    this.semaphoreOneway.availablePermits()//
                );
                logger.warn(info);
                throw new RemotingTimeoutException(info);
            }
        }

    }

    public void invokeAsyncImpl(final Channel channel, final RemotingCommand request,
        final long timeoutMillis, final InvokeCallback invokeCallback) throws InterruptedException,
        RemotingTooMuchRequestException, RemotingTimeoutException, RemotingSendRequestException {
        final int opaque = request.getOpaque();// 每次请求的唯一标识
        boolean acquired = this.semaphoreAsync.tryAcquire(timeoutMillis, TimeUnit.MILLISECONDS);
        if (acquired) {
            final SemaphoreReleaseOnlyOnce once = new SemaphoreReleaseOnlyOnce(this.semaphoreAsync);

            final ResponseFuture responseFuture =
                new ResponseFuture(opaque, timeoutMillis, invokeCallback, once);
            this.responseTable.put(opaque, responseFuture);
            try {
                channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture f) throws Exception {
                        if (f.isSuccess()) {
                            responseFuture.setSendRequestOK(true);// 如果发送成功返回
                            return;
                        } else {// 发送失败从responseTable 移除并回调
                            responseFuture.setSendRequestOK(false);
                            responseFuture.putResponse(null);
                            responseTable.remove(opaque);
                            try {
                                executeInvokeCallback(responseFuture);// 失败回调
                            } catch (Throwable e) {
                                logger.warn(
                                    "excute callback in writeAndFlush addListener, and callback throw",
                                    e);
                            } finally {
                                responseFuture.release();
                            }
                            logger.warn("send a request command to channel <{}> failed.",
                                RemotingHelper.parseChannelRemoteAddr(channel));
                        }

                    }
                });
            } catch (Exception e) {
                responseFuture.release();
                logger.warn("send a request command to channel <"
                    + RemotingHelper.parseChannelRemoteAddr(channel) + "> Exception", e);
                throw new RemotingSendRequestException(
                    RemotingHelper.parseChannelRemoteAddr(channel), e);
            }
        } else {
            String info = String.format(
                "invokeAsyncImpl tryAcquire semaphore timeout, %dms, waiting thread nums: %d semaphoreAsyncValue: %d",
                //
                timeoutMillis, //
                this.semaphoreAsync.getQueueLength(), //
                this.semaphoreAsync.availablePermits()//
            );
            logger.warn(info);
            throw new RemotingTooMuchRequestException(info);
        }
    }

    public void processRequestCommand(ChannelHandlerContext ctx, RemotingCommand cmd) {
        // 每一个NettyRequestProcessor对应一个线程池
        final Pair<NettyRequestProcessor, ExecutorService> matched =
            this.processorTable.get(cmd.getCode());

        final Pair<NettyRequestProcessor, ExecutorService> pair =
            null == matched ? this.defaultRequestProcessor : matched;

        final int opaque = cmd.getOpaque();// 每个请求的唯一编码

        if (null != pair) {
            // 拒绝请求
            if (pair.getT1().rejectRequest()) {
                final RemotingCommand response =
                    RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                        "[REJECTREQUEST]system busy, start flow control for a while");
                response.setOpaque(opaque);
                ctx.writeAndFlush(response);
                return;
            }
            Runnable requestRunable = buildRequestTaskRunnable(ctx, cmd, pair, opaque);
            try {
                pair.getT2().submit(requestRunable);
            } catch (RejectedExecutionException e) {
                if ((System.currentTimeMillis() % 10000) == 0) {
                    logger.warn(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) //
                        + ", too many requests and system thread pool busy, RejectedExecutionException "
                        //
                        + pair.getT2().toString() //
                        + " request code: " + cmd.getCode());
                }

                if (!cmd.isOnewayRPC()) {
                    final RemotingCommand response =
                        RemotingCommand.createResponseCommand(RemotingSysResponseCode.SYSTEM_BUSY,
                            "[OVERLOAD]system busy, start flow control for a while");
                    response.setOpaque(opaque);
                    ctx.writeAndFlush(response);
                }
            }
        } else {
            String error = " request type " + cmd.getCode() + " not supported";
            final RemotingCommand response = RemotingCommand
                .createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
            response.setOpaque(opaque);
            ctx.writeAndFlush(response);
            logger.error(RemotingHelper.parseChannelRemoteAddr(ctx.channel()) + error);
        }

    }

    private Runnable buildRequestTaskRunnable(ChannelHandlerContext ctx, RemotingCommand cmd,
        final Pair<NettyRequestProcessor, ExecutorService> pair, final int opaque) {
        Runnable run = new Runnable() {
            @Override
            public void run() {
                try {
                    RPCHook hook = NettyRemotingAbstract.this.getRPCHook();// 获取注册的钩子函数

                    if (null != hook) {
                        hook.doBeforeRequest(RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                            cmd);
                    }

                    final RemotingCommand response = pair.getT1().processRequest(ctx,
                        cmd);// 执行注册的请求code对应的NettyRequestProcessor获取给client的回应结果

                    if (hook != null) {
                        hook.doAfterResponse(RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                            cmd,
                            response);
                    }

                    /**
                     * 服务器返回给客户端body压缩待做 TODO 压缩response
                     *
                     * int sysFlag = 0; if (ClientUtils.tryToCompressMessage(response)) {// 根据压缩配置执行消息压缩。
                     * sysFlag |= ClientUtils.COMPRESSED_FLAG; } response.addExtField("sysFlag", sysFlag +
                     * "");
                     */

                    // 如果是不是单程调用
                    if (!cmd.isOnewayRPC()) {
                        if (response != null) {
                            response.setOpaque(opaque);// 给response设置request同一唯一标识
                            response.markResponseType();// 转换为为response
                            try {
                                response.setCode(RemotingSysResponseCode.SUCCESS);// 回调函数成功
                                ctx.writeAndFlush(response);
                            } catch (Throwable e) {
                                logger.error("process request over, but response failed", e);
                                logger.error(cmd.toString());
                                logger.error(response.toString());
                            }
                        } else {

                        }
                    }
                } catch (Throwable e) {
                    logger.debug(e.getMessage());
                    if (!cmd.isOnewayRPC()) {
                        final RemotingCommand response = RemotingCommand.createResponseCommand(
                            RemotingSysResponseCode.SYSTEM_ERROR,
                            RemotingHelper.exceptionSimpleDesc(e));
                        response.setOpaque(opaque);
                        ctx.writeAndFlush(response);
                    }
                }
            }

        };
        return run;
    }

    class NettyEventExecuter extends ServiceThread {

        private final LinkedBlockingQueue<NettyEvent> eventQueue =
            new LinkedBlockingQueue<>();
        private final int maxSize = 10000;

        public void putNettyEvent(final NettyEvent event) {
            if (this.eventQueue.size() <= maxSize) {
                this.eventQueue.add(event);
            } else {
                logger.warn("event queue size[{}] enough, so drop this event {}",
                    this.eventQueue.size(),
                    event.toString());
            }
        }

        @Override
        public void run() {
            logger.info(this.getServiceName() + " service started");
            final ChannelEventListener listener = NettyRemotingAbstract.this
                .getChannelEventListener();
            while (!this.isStopped()) {
                try {
                    NettyEvent event = this.eventQueue.poll(3000, TimeUnit.MILLISECONDS);
                    if (event != null && listener != null) {
                        switch (event.getType()) {
                            case IDLE:
                                listener.onChannelIdle(event.getRemoteAddr(), event.getChannel());
                                break;
                            case CLOSE:
                                listener.onChannelClose(event.getRemoteAddr(), event.getChannel());
                                break;
                            case CONNECT:
                                listener
                                    .onChannelConnect(event.getRemoteAddr(), event.getChannel());
                                break;
                            case EXCEPTION:
                                listener
                                    .onChannelException(event.getRemoteAddr(), event.getChannel());
                                break;
                            default:
                                break;

                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

        @Override
        public String getServiceName() {
            return NettyEventExecuter.class.getSimpleName();
        }

    }
}


